package com.scaffold.commons.file;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

/**
 * 文件服务的hadoop实现
 * @author wangjg
 *
 */
public class HDFileSystem implements FileIO {
	private String hdfsUrl;
	private FileSystem fs ;
	
	private FileSystem fs() throws IOException {
		if(fs==null){
			synchronized(this){
				if(fs==null){
					Configuration conf = new Configuration();
				    this.fs = FileSystem.get(URI.create(hdfsUrl), conf);
			    }
			}
		}
		return fs;
	}
	
	public HDFileSystem(){
		
	}
	
	public void setHadoopUserName(String username){
		System.setProperty("HADOOP_USER_NAME", username);
	}
	
	@Override
	public void save(String path, InputStream in) throws IOException{
	    Path remoteFile = new Path(path);
	    OutputStream out = this.fs().create(remoteFile);
	    try{
	    	IOUtils.copyBytes(in, out, 1024,true);
	    }finally{
	    	if(out!=null)out.close();
	    }
	}
	
	@Override
	public void load(String path, OutputStream out) throws IOException {
	    InputStream in = null;
	    try{
	    	in = this.getInputStream(path);
	    	IOUtils.copyBytes(in, out, 1024,true);
	    }finally{
	    	if(in!=null)in.close();
	    }
	}
	
	@Override
	public InputStream getInputStream(String path) throws IOException {
	    Path remoteFile = new Path(path);
	    FSDataInputStream in = this.fs().open(remoteFile);
	    return in;
	}
	
	@Override
	public OutputStream getOutputStream(String path) throws IOException {
	    Path remoteFile = new Path(path);
	    FSDataOutputStream out = this.fs().create(remoteFile, true);
	    return out;
	}
	
	@Override
	public boolean exists(String path) throws IOException {
	    Path remoteFile = new Path(path);
	    return this.fs().exists(remoteFile);
	}
	
	public boolean delete(String path) throws IOException {
	    Path remoteFile = new Path(path);
	    return this.fs().delete(remoteFile, true);
	}

	@Override
	public boolean move(String srcPath, String destPath) throws IOException {
		return this.fs().rename(new Path(srcPath), new Path(destPath));
	}

	public String getHdfsUrl() {
		return hdfsUrl;
	}

	public void setHdfsUrl(String hdfsUrl) {
		this.hdfsUrl = hdfsUrl;
	}

	public long getLength(String path) throws IOException {
	    Path remoteFile = new Path(path);
	    FileStatus fStatus = this.fs().getFileStatus(remoteFile);
		return fStatus.getLen();
	}
	
	@Override
	public List<String> getChildren(String path) throws IOException{
		List<String> list = new ArrayList<String>();
		Path remoteFile = new Path(path);
		FileStatus[] fss = this.fs().listStatus(remoteFile);
		for(FileStatus s:fss){
			list.add(s.getPath().getName());
		}
		return list;
	}
	
	@Override
	public FileDetail getFileStatus(String path) throws IOException{
		Path remoteFile = new Path(path);
		FileStatus status = this.fs().getFileStatus(remoteFile);
		FileDetail fd = new FileDetail();
		fd.setDir(status.isDir());
		fd.setSize(status.getLen());
		return fd;
	}
	
	public void test(String path) throws IOException{
		Path remoteFile = new Path(path);
		this.fs().mkdirs(remoteFile);
	}
	
	public static void main(String args[]) throws IOException{
//		String hdfsUrl = "hdfs://192.168.2.183/";
		String hdfsUrl = "hdfs://192.168.25.129:9000/";
		HDFileSystem fs = new HDFileSystem();
		fs.setHdfsUrl(hdfsUrl);
//		fs.setHdfsUrl(hdfsUrl);
		List<String> list = fs.getChildren("/");
		System.out.println("result:"+list);
		
	}
}
